package com.niodata.dp.task;

import com.niodata.dp.task.entity.PackageParam;
import com.niodata.dp.util.JsonUtil;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.commons.io.IOUtils;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;

public class JobTypeParamSchemas {

  private static Map<String, List<PackageParam>> paramSchemas = new HashMap<>();
  private static Set<String> jobTypeInternalParams = new HashSet<String>();

  /**
   * load job type param schemas from class path.
   */
  static {
    String[] internalParams = new String[]{"type", "keytab", "principal",
        "dependencies", "working.dir", "name"};
    for (String p : internalParams) {
      jobTypeInternalParams.add(p);
    }
    PathMatchingResourcePatternResolver resourcePatternResolver =
        new PathMatchingResourcePatternResolver();
    try {
      Resource[] resources = resourcePatternResolver
          .getResources("classpath:META-INF/param.schemas/v1/*");
      for (Resource res : resources) {
        String fileName = res.getFilename();
        String[] ss = fileName.split("/");
        String name = ss[ss.length - 1];
        name = name.split("\\.")[0];
        InputStream inputStream = res.getInputStream();
        String json = new String(IOUtils.toByteArray(inputStream));
        List<PackageParam> schemas = JsonUtil.fromJsonList(PackageParam.class, json);
        paramSchemas.put(name, schemas);
        inputStream.close();
      }
    } catch (IOException ioe) {
      throw new RuntimeException(ioe);
    }
  }

  private JobTypeParamSchemas() {
  }

  public static List<PackageParam> getJobTypeParamSchemas(String jobtype, String language) {
    List<PackageParam> list = new ArrayList<>();
    /*
    String key = jobtype + "-" + language;
    if (jobtype.equals("command")) {
      key = jobtype;
    }
    */
    String key = jobtype;
    if (jobtype.equals("spark")) {
      key = jobtype + "-" + language;
    }
    list.addAll(paramSchemas.get(key));
    return list;
  }

  public static boolean isUserSetAble(String jobType, String language, String name) {
    for (PackageParam param : getJobTypeParamSchemas(jobType, language)) {
      if (param.getName().equals(name) && param.isUserSetAble()) {
        return true;
      }
    }
    return false;
  }

  public static Map<String, String> getJobTypeDefaultParams(String jobtype, String language) {
    List<PackageParam> list = getJobTypeParamSchemas(jobtype, language);
    Map<String, String> map = new HashMap<>();
    for (PackageParam schema : list) {
      if (schema.getDefaultValue() != null && schema.getDefaultValue().length() > 0) {
        map.put(schema.getName(), schema.getDefaultValue());
      }
    }
    return map;
  }

  public static boolean isInternalParam(String name) {
    return jobTypeInternalParams.contains(name);
  }

}
